package storm2;// -*- codeing: utf-8 -*-
// @Time     :2021/12/9 0:37
// @Author   :郭荣杰
//@File      :Bolt1.java
//@Software  : IntelliJ IDEA
import net.sf.json.JSONArray;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import net.sf.json.JSONObject;
public class Bolt1_order extends BaseBasicBolt{

    public void execute(Tuple input, BasicOutputCollector collector) {
        // TODO Auto-generated method stub
        /**
         * kafkaSpout发送的字段名为bytes
         */

        /**
         * 描述字段
         */
        /**-----------------------------------------------
        /**
         * 1.order_id	    varchar(50)    订单唯一编号
         * 2.book_id	    varchar(50)    图书ID
         * 3.book_name	    varchar(255)   图书名称
         * 4.book_type	    varchar(10)    图书类型
         * 5.book_num	    int            图书数量
         * 6.book_price	    float          图书单位(多少元)
         * 7.book_press	    varchar(255)   出版社
         * 8.id	            int            订单唯一 ID 订单自增长 ID
         */
        try{
            //得到字节码
            byte[]bytes=input.getBinary(0);  //字节码
            //byte->String
            //192.168.53.123	0-001	100	手机	小米8	99	2	20181112（timeStamp）
            //字节码转字符串
            String line =new String(bytes);
            //声明一个JSONObejct对象
            JSONObject jsonObject=new net.sf.json.JSONObject();
            //声明一个JSONArray对象
            JSONArray jsonArray =new JSONArray();
            //利用jsonObject方法将字符串转JSON
            JSONObject json_line = jsonObject.fromObject(line);
            //得到数据data，去除元数据
            JSONArray data =jsonArray.fromObject(json_line.getString("data"));
            //得到数据data各字段的值
//        System.out.println(data.toArray()[0]);
//            String order_id =jsonObject.fromObject(data.toArray()[0]).getString("order_id");
//            String book_id=jsonObject.fromObject(data.toArray()[0]).getString("book_id");
            String book_name =jsonObject.fromObject(data.toArray()[0]).getString("book_name");
            String book_type =jsonObject.fromObject(data.toArray()[0]).getString("book_type");
            int book_num=jsonObject.fromObject(data.toArray()[0]).getInt("book_num");
//            double book_price=jsonObject.fromObject(data.toArray()[0]).getDouble("book_price");
            String book_press=jsonObject.fromObject(data.toArray()[0]).getString("book_press");
//            int id=jsonObject.fromObject(data.toArray()[0]).getInt("id");
//            collector.emit(new Values(order_id,book_id,book_name,book_type,book_num,book_price,book_press,id));
            collector.emit(new Values(book_name,book_type,book_num,book_press));
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        /**
         * 描述字段
         */
        /**-----------------------------------------------
         /**
         * 1.order_id	    varchar(50)    订单唯一编号
         * 2.book_id	    varchar(50)    图书ID
         * 3.book_name	    varchar(255)   图书名称
         * 4.book_type	    varchar(10)    图书类型
         * 5.book_num	    int            图书数量
         * 6.book_price	    float          图书单位(多少元)
         * 7.book_press	    varchar(255)   出版社
         * 8.id	            int            订单唯一 ID 订单自增长 ID
         */
//        declarer.declare(new Fields("order_id", "book_id", "book_name","book_type","book_num","book_price",
//                "book_press","id"));

        declarer.declare(new Fields("book_name","book_type","book_num","book_press"));
    }

}
